package cn.ac.iscas.service.flink;


//import cn.ac.iscas.dao.api.ComponentMetadataRepository;
//import cn.ac.iscas.dao.api.JobMetadataRepository;
import cn.ac.iscas.dao.IComponentMetadataDao;
import cn.ac.iscas.dao.IJobMetadataDao;
import cn.ac.iscas.dao.entity.ComponentMetadataDO;
import cn.ac.iscas.domain.common.EtlColumn;
import cn.ac.iscas.domain.component.reader.ReaderParameter;
import cn.ac.iscas.domain.component.transformer.TransformerParameter;
import cn.ac.iscas.domain.job.*;
import cn.ac.iscas.domain.response.AnalyzeColumResponse;
import cn.ac.iscas.domain.response.job.JobSearchResponse;
import cn.ac.iscas.domain.view.Option;
import cn.ac.iscas.service.common.JobExecutor;
import cn.ac.iscas.service.common.IJobService;
import cn.ac.iscas.service.common.analyzer.AnalyzerParameter;
import cn.ac.iscas.service.quartz.QuartzJobService;
import cn.ac.iscas.utils.ClassLoaderSwapper;
import cn.ac.iscas.utils.DataProcessUtils;
import cn.ac.iscas.utils.JarLoader;
import com.fasterxml.jackson.core.type.TypeReference;
import com.iscas.datasong.lib.common.DataSongException;
import com.iscas.datasong.lib.common.Status;
import com.iscas.datasong.lib.common.column.ColumnType;
import com.iscas.datasong.lib.util.DataSongExceptionUtils;
import com.iscas.datasong.lib.util.DataSongJsonUtils;
import com.iscas.datasong.lib.util.DataSongStringUtils;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.util.*;

/**
 * @description:
 * @param: null
 * @return:
 * @author LJian
 * @date: 2022/11/15 8:47
 */
@Service
public class SqlJobService implements IJobService {
    ///@Autowired
    ///private EtlJobDao etlJobDao;
    ///@Autowired
	///private JobMetadataRepository jobMetadataRepository;
    @Autowired
	JobExecutor jobExecutor;
    @Autowired
	QuartzJobService quartzJobExecuteService;

	@Autowired
	IComponentMetadataDao componentMetadataDao;

	@Autowired
	//private JobMetadataDao jobMetadataDao;
	IJobMetadataDao jobMetadataDao;

    /**
     * 查询任务列表
     *
     * @param jobSearch
     * @return
     */
    @Override
    public JobSearchResponse searchJob(JobSearch jobSearch) throws DataSongException {
		JobSearchResponse response = new JobSearchResponse();
		//List<Job> jobs = jobMetadataDao.select(jobSearch);
		List<Job> jobs = jobMetadataDao.selectByGroup(jobSearch);

		response.setJobs(jobs);
		response.setTotal(jobs.size());
		return response;
    }

	public Map<String, String>  getAllJob( ) throws DataSongException {
		///JobSearchResponse response =  etlJobDao.getAllJob();
		//JobSearchResponse response = jobMetadataDao.getAllJob();
		List<Job> jobs = jobMetadataDao.selectAll();
		Map<String, String> map = new HashMap<>();
		for(Job job : jobs){
			map.put(job.get_id(), job.getName());
		}
		return map;
	}

    /**
     * 根据id删除任务 delete
     *
     * @param id 任务唯一id
     * @return 返回结果
     */
    @Override
    public boolean deleteJob(String id) throws DataSongException {
		return jobMetadataDao.delete(id);
    }

	@Override
	public boolean deleteJob(List<String> ids) throws DataSongException {
		return jobMetadataDao.deleteBatch(ids);
	}

	/**
     * 根据id号获取一个job的详情
     *
     * @param id 任务唯一id
     * @return
     */
    @Override
    public Job getJob(String id) throws DataSongException {
        return jobMetadataDao.getJob(id);
    }


	public List<Option> getColumnType() throws DataSongException {
		List<Option> options = new ArrayList<>();
    	for(ColumnType type :  ColumnType.values()){
    		Option option = new Option(type.name(), type.name());

			options.add(option);
		}

		return options;
	}


	/**
     * 任务校验 post
     *
     * @param job
     * @return
     */
    @Override
    public boolean validateJob(Job job) throws DataSongException {
        ///List<Component> components =  job.buidObjJobContent(etlConfigure.getDataSongFileStreamServiceAddr(DataSongAnnotationUtils.getTableName(Job.class)));
        ///List<Component> components =  job.buidObjJobContent();
        List<Component> components =  job.getComponents();
        if(components == null || components.size()==0){
			throw new DataSongException(Status.PARAM_ERROR, "任务不能为空！");
		}

        for (int i = 0; i < components.size(); i++) {
            Component component = components.get(i);
            if(!component.getParameter().validate()){
            	throw new DataSongException(Status.PARAM_ERROR, String.format("component [%s] validate error",component.getAlias()));
			}
        }
        return true;
    }

     /**
     * 任务执行 post
     *
     * @param job
     * @return
     */
	 @Override
    public String executeJob(Job job) throws DataSongException {
        //校验Job
        validateJob(job);

        //检测Job状态
		Job jobMetadata = null;
		if(DataSongStringUtils.isNotEmpty(job.get_id())) {
			///jobMetadata = etlJobDao.getJob(job.get_id());
			jobMetadata = jobMetadataDao.getJob(job.get_id());
			if(jobMetadata != null) {
				JobStatus jobStatus = JobStatus.analyze(jobMetadata.getStatus());
				if (jobStatus.isRunning()) {
					throw new DataSongException(Status.PARAM_ERROR, "the job is running, can not run again!");
				}
				job.setStatus(jobMetadata.getStatus());
			}
		}else{
			//尚未保存过，设置为默认状态
			job.setStatus(JobStatus.UNSUBMITTED.name());
		}

		//检查参数是否具备
		List<Component> components = (List<Component>) job.getContent();
		///List<ComponentMetadataDO> componentMetadataList = componentMetadataRepository.findAll();
		List<ComponentMetadataDO> componentMetadataList = componentMetadataDao.selectAll();
		for(Component component : components){
			if(component.getType() == ComponentType.analyzer){
				ComponentMetadataDO componentMetadata = null;
				for(ComponentMetadataDO tmp : componentMetadataList){
					if(component.getEntryClassName().equals(tmp.getEntryClassName())){
						componentMetadata = tmp;
						break;
					}
				}
				if(componentMetadata == null){
					throw new DataSongException(Status.PARAM_ERROR, String.format("找不到name=[%s]的组件",component.getEntryClassName()));
				}
				//TODO：2023-01-29 改成HDFS存储
//				AnalyzerParameter analyzerParameter = (AnalyzerParameter) component.getParameter();
//				if(DataSongStringUtils.isNotEmpty(componentMetadata.getModelLocalPath())){
//					if (componentMetadata.getModelLocalPath().startsWith("http")) {
//						if (DataSongStringUtils.isNotEmpty(componentMetadata.getModelPublishPath())
//								&& etlFileDao.modelFileExist(componentMetadata.getModelPublishPath())) {
//							//do nothing
//						} else {
//							String fileId = etlFileDao.uploadModelFile(componentMetadata.getModelLocalPath());
//							componentMetadata.setModelPublishPath(fileId);
//							///componentMetadataRepository.save(componentMetadata);
//							componentMetadataDao.save(componentMetadata);
//						}
//						///analyzerParameter.setModelPath(etlConfigure.getDataSongFileStreamServiceAddr(DataSongAnnotationUtils.getTableName(ComponentMetadataDO.class),
//						///		componentMetadata.getModelPublishPath()));
//					} else if (componentMetadata.getModelLocalPath().startsWith("hdfs")) {
//						analyzerParameter.setModelPath(componentMetadata.getModelLocalPath());
//					}
//				}
				componentMetadataDao.save(componentMetadata);
			}
		}


		//尚未保存过，先保存
		String jobId = saveJob(job);
		job.set_id(jobId);


         //提交执行Job
		quartzJobExecuteService.submitJob(job);

        return jobId;
    }

	/**
	 * 任务执行 post
	 *
	 * @param jobIds
	 * @return
	 */
	@Override
	public boolean executeJob(List<String> jobIds) throws DataSongException {
		//检测Job状态
		///List<Job> jobs = jobMetadataRepository.findAll(jobIds);
		List<Job> jobs = jobMetadataDao.selectByIds(jobIds);
		for(Job job : jobs) {
			JobStatus jobStatus = JobStatus.analyze(job.getStatus());
			if (jobStatus.isRunning()) {
				throw new DataSongException(Status.PARAM_ERROR, "the job is running, can not run again!");
			}
		}

		//提交执行Job
		for(Job job : jobs) {
//			validateJob(job);
//			quartzJobExecuteService.submitJob(job);
			executeJob(job);
		}

		return true;
	}

	@Override
	public boolean stopJob(List<String> jobIds) throws DataSongException {
		//检测Job状态
		///List<Job> jobs = jobMetadataRepository.findAll(jobIds);
		List<Job> jobs = jobMetadataDao.selectByIds(jobIds);
		for(Job job : jobs) {
			JobStatus jobStatus = JobStatus.analyze(job.getStatus());
			if (jobStatus.isRunning()) {
				quartzJobExecuteService.stop(job);
				///job.setStatus(JobStatus.PAUSE.name()));
				///jobMetadataRepository.save(job);
				jobMetadataDao.updateJobStatus(job.get_id(), JobStatus.PAUSE.name());
			} else {
				//todo:Pause?
				//jobMetadataDao.updateJobStatus(job.get_id(), jobStatus.name());
				jobMetadataDao.updateJobStatus(job.get_id(), JobStatus.PAUSE.name());
			}
		}

		return true;
	}

    public SqlJobService() {
        super();
    }

    /**
     * 获取Job日志
     *
     * @param id
     * @return
     * @throws DataSongException
     */
    @Override
    public List<JobLog> getJobLog(String id) throws DataSongException {
        List<JobLog> jobLogs = jobExecutor.getJobLog(id);
        ///Job jobMetadata = etlJobDao.getJob(id);
        Job jobMetadata = jobMetadataDao.getJob(id);
        String preStatus = jobMetadata.getStatus();
        jobMetadata.set_id(id);
        if (jobLogs.size() > 0) {
            String jobStatus = jobLogs.get(jobLogs.size() - 1).getStatus().name();
            if (!preStatus.equals(jobStatus)) {
                jobMetadata.setStatus(jobStatus);
                ///etlJobDao.save(jobMetadata);
                jobMetadataDao.save(jobMetadata);
            }
            if (jobLogs.get(jobLogs.size() - 1).getStatus().isFinished()) {
                jobMetadata.setDuration(new Date().getTime() - jobMetadata.getExecuteTime().getTime());
                ///etlJobDao.save(jobMetadata);
                jobMetadataDao.save(jobMetadata);
            }
        }
        return jobLogs;
    }

    /**
     * 任务保存 post
     *
     * @return 返回job的唯一id
     */
    @Override
    public boolean updateJobStatus(String id, JobStatus jobStatus) {
        ///Job jobMetadata = etlJobDao.getJob(id);
//        Job jobMetadata = jobMetadataDao.getJob(id);
//        String preStatus = jobMetadata.getStatus();
//        jobMetadata.set_id(id);
//        if (!jobStatus.name().equals(preStatus)) {
//            jobMetadata.setStatus(jobStatus.name());
//			jobMetadataDao.updateJobStatus(id, jobStatus.name());
//        }
		jobMetadataDao.updateJobStatus(id, jobStatus.name());
        return true;
    }

    /**
	 * 任务保存 post
	 *
	 * @param job
	 * @return 返回job的唯一id
	 */
	@Override
	public String saveJob(Job job) throws DataSongException {
		String jobId = job.get_id();
		if(jobId != null && !jobId.trim().isEmpty()){
			///Job tmp = etlJobDao.getJob(job.get_id());
			Job tmp = jobMetadataDao.getJob(jobId);
			if(tmp != null) {
				JobStatus jobStatus = JobStatus.analyze(tmp.getStatus());
				if (jobStatus.isRunning()) {
					throw new DataSongException(Status.PARAM_ERROR, "Job is ruuning, can not be updated!");
				}
				job.setStatus(jobStatus.name());
			}
		}else{
			job.setStatus(JobStatus.UNSUBMITTED.name());
			job.setCreateTime(Date.from(ZonedDateTime.now(ZoneId.of("Asia/Shanghai")).toInstant()));
			jobId = UUID.randomUUID().toString().replace("-", "");
			job.set_id(jobId);
		}


		//置空之前的运行状态
		job.setInnerId(null);
		job.setRunInfo(null);

		if(job.getCreateTime() == null){
			job.setCreateTime(Date.from(ZonedDateTime.now(ZoneId.of("Asia/Shanghai")).toInstant()));
		}

		///return etlJobDao.save(job);
		int res = jobMetadataDao.save(job);
		if (res != 1) {
			jobId = null;
		}
		return jobId;
	}

	/**
	 * 任务保存 post
	 *
	 * @param jobs
	 * @return 返回job的唯一id
	 */
	@Override
	public List<String> saveJob(List<Job> jobs)  {
		List<String> ids = new ArrayList<>();
		for(Job job : jobs) {
			String jobId = UUID.randomUUID().toString().replace("-", "");
			job.set_id(jobId);  //批量保存，相当于新建，id重新自动生成

			//置空之前的运行状态
			job.setInnerId(null);
			job.setRunInfo(null);
			job.setCreateTime(Date.from(ZonedDateTime.now(ZoneId.of("Asia/Shanghai")).toInstant()));

			job.setStatus(JobStatus.UNSUBMITTED.name());

//			job.buidStrJobContent();
			///ids.add(jobMetadataRepository.save(job));
			if (1 == jobMetadataDao.save(job)) {
				ids.add(jobId);
			}
		}

		return ids;
	}

    /**
     * 任务清空
     *
     * @param id
     * @return
     */
    @Override
    public String clearJob(String id) throws DataSongException {
        ///Job jobMetadata = etlJobDao.getJob(id);
        Job jobMetadata = jobMetadataDao.getJob(id);

        if(jobMetadata == null){
        	throw new DataSongException(Status.PARAM_ERROR, String.format("job [%s] not exist!", id));
		}

        if(JobStatus.analyze(jobMetadata.getStatus()).isRunning()){
			throw new DataSongException(Status.PARAM_ERROR, "job is running, can not be cleared!");
		}

        jobMetadata.setContent(null);

        ///return etlJobDao.save(jobMetadata);
		int res = jobMetadataDao.update(jobMetadata);
		if (res != 1) {
			return null;
		}
        return id;
    }

	public AnalyzeColumResponse analyzeComponentColumn(Job job, String cId) throws DataSongException {
		///job.buidObjJobContent(etlConfigure.getDataSongFileStreamServiceAddr(DataSongAnnotationUtils.getTableName(Job.class)));
		///job.buidObjJobContent();
		List<Component> components = (List<Component>) job.getContent();

		//找根节点，暂时只处理一个根节点的情况
		Component component = null;
		for (int i = 0; i < components.size(); i++) {
			if(components.get(i).getId().equals(cId)){
				component = components.get(i);
				break;
			}
		}

		if(component == null){
			throw new DataSongException(Status.PARAM_ERROR, String.format("任务中不存在与[%s]对应的组件！",cId));
		}

		if (component.getType() == ComponentType.reader) {
			return analyzeInitialColumn(component);
		}else{
			return analyzeInputColumn(components, cId);
		}

	}
    /**
     * 获取初始列字段名
     *
     * @param component
     * @return
     * @throws DataSongException
     */

    private AnalyzeColumResponse analyzeInitialColumn(Component component) throws DataSongException {
		if(!component.getParameter().validate()){
			throw new DataSongException(Status.PARAM_ERROR, "参数错误！");
		}

		AnalyzeColumResponse readerColumResponse = new AnalyzeColumResponse();

		ReaderParameter readerParameter = (ReaderParameter) component.getParameter();
		readerParameter.validate();

		Map simpleParameter = readerParameter.getSimpleParameter();

		String fullClassName = component.getEntryClassName();
		int index = fullClassName.lastIndexOf(".");
		if(index<0){
			throw new DataSongException(Status.CONFIG_INFO_ERROR, String.format("无效的组件配置信息，name=%s", fullClassName));
		}
		String jarName = fullClassName.substring(index+1);
		//load jar
		try {
			//todo:打包部署以后要通过配置文件拿到模型文件路径
			String path = System.getProperty("user.dir");
			JarLoader jarLoader = null;
			if(path.contains("ig-bootstrap")){
				path = path + "/src/main/resources/component/"+jarName+".jar";
			}else{
				path = path + "/ig-bootstrap/src/main/resources/component/"+jarName+".jar";
			}
			jarLoader = new JarLoader(new String[]{path});
			ClassLoaderSwapper classLoaderSwapper = ClassLoaderSwapper.newCurrentThreadClassLoaderSwapper();
			classLoaderSwapper.setCurrentThreadClassLoader(jarLoader);
			Class<?> adClass = Thread.currentThread().getContextClassLoader().loadClass(component.getEntryClassName());

			Object readerComponent =   adClass.newInstance();

			//初始化
			Method initMethod = adClass.getDeclaredMethod("init", Map.class);
			initMethod.invoke(readerComponent, simpleParameter);
			initMethod = null;

			//获取表头
			Method getHeaderMethod = adClass.getDeclaredMethod("getHeader");
			List<EtlColumn>  columns = (List<EtlColumn>) getHeaderMethod.invoke(readerComponent);
			getHeaderMethod = null;

			//获取数据
			Method getSampleDataMethod = adClass.getDeclaredMethod("getSampleData", int.class);
			List<Map<String,Object>>  datas = (List<Map<String,Object>>) getSampleDataMethod.invoke(readerComponent, 10);
			getSampleDataMethod = null;
			readerComponent = null;

			classLoaderSwapper.restoreCurrentThreadClassLoader();

			readerColumResponse.setHeader(columns);
			readerColumResponse.setDatas(datas);
		}catch (Exception e){
			e.printStackTrace();
			if(e instanceof  DataSongException){
				throw  (DataSongException)e;
			}else if(e instanceof InvocationTargetException  ){
				throw new DataSongException(Status.SERVER_ERROR, "加载reader组件错误：" + DataSongExceptionUtils.getExceptionInfo(((InvocationTargetException) e).getTargetException()));
			}else  {
				throw new DataSongException(Status.SERVER_ERROR, "加载reader组件错误：" + DataSongExceptionUtils.getExceptionInfo(e));
			}
		}

		return readerColumResponse;
    }

    /**
     * 获取组件数据输入列
     *
     * @param components
     * @param cId 组件编id
     * @return
     */
    private AnalyzeColumResponse analyzeInputColumn(List<Component> components, String cId) throws DataSongException {
        List<EtlColumn> componentColumns = new ArrayList<>();
        Component component = null;

        //找根节点，暂时只处理一个根节点的情况
        for (int i = 0; i < components.size(); i++) {
            if (components.get(i).getFront().isEmpty() && components.get(i).getNext()!=null
				&& components.get(i).getNext().size()>0) {
                component = components.get(i);
                break;
            }
        }

		List<String> columnNames = null;
        if(component != null ) {
        	//获取输入源的列信息
			ReaderParameter readerParameter = (ReaderParameter) component.getParameter();
			if (readerParameter.getColumns() == null) {
				Object tmpEtlColumns = analyzeInitialColumn(component).getHeader();//不同类加载器加载的同一个类也相互不认识，需要做一次转换
				List<EtlColumn> etlColumns = DataSongJsonUtils.fromJson(DataSongJsonUtils.toJson(tmpEtlColumns),new TypeReference<List<EtlColumn>>(){});
				readerParameter.setColumns(etlColumns);
			}
			componentColumns = readerParameter.getColumns();
			columnNames = DataProcessUtils.transformEtlColumn(componentColumns);

			List<String> componentNextIds = new ArrayList<>();
			while (true) {
				componentNextIds = component.getNext();
				//串行
				if (componentNextIds.size() == 1) {
					for (int i = 0; i < components.size(); i++) {
						if (componentNextIds.get(0).equals(components.get(i).getId())) {
							//下一个节点赋予component
							component = components.get(i);
							break;
						}
					}

					if (component.getId().equals(cId)) {
						break;
					}

					component.getParameter().validate();

					switch (component.getType() ) {
						case transformer:
							TransformerParameter transformerParameter = (TransformerParameter) component.getParameter();
							if (componentColumns.size() > 0) {
								columnNames = transformerParameter.analyzeOutputColumn(columnNames);
							} else {
								throw new DataSongException(Status.PARAM_ERROR, component.getId() + "is null");
							}
							break;
						case analyzer:
							AnalyzerParameter parameter = (AnalyzerParameter) component.getParameter();
							if (componentColumns.size() > 0) {
								columnNames = parameter.analyzeOutputColumn(columnNames);
							} else {
								throw new DataSongException(Status.PARAM_ERROR, component.getId() + "is null");
							}
							break;
					}
				}else{
					throw new DataSongException(Status.SERVER_ERROR,"暂不支持多分支");
				}
			}
		}

        //转换ComponentColumn
		List<EtlColumn> results = new ArrayList<>();
		for (int i = 0; columnNames  != null && i < columnNames.size(); i++) {
			for(EtlColumn etlColumn : componentColumns) {
				if(etlColumn.getName().equals(columnNames.get(i))) {
					results.add(etlColumn);
				}
			}
		}

		AnalyzeColumResponse response = new AnalyzeColumResponse();
		response.setHeader(results);

		return response;
    }

}
